package com.new1cloud.file.upload.client.bootstrap;

import com.new1cloud.file.domain.FileChunkInfo;
import com.new1cloud.file.domain.FileUploadBasicInfo;
import com.new1cloud.file.domain.UploadProgressListener;
import com.new1cloud.file.tools.FileHashGenerate;
import com.new1cloud.file.upload.client.bootstrap.handler.FileUploadClientHandler;
import com.new1cloud.file.upload.client.bootstrap.handler.LocalCache;
import com.new1cloud.file.upload.client.entry.ChunkFile;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import io.netty.handler.codec.LengthFieldPrepender;
import io.netty.handler.codec.serialization.ClassResolvers;
import io.netty.handler.codec.serialization.ObjectDecoder;
import io.netty.handler.codec.serialization.ObjectEncoder;
import lombok.Setter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.File;
import java.util.*;
import java.util.concurrent.*;
import java.util.stream.Collectors;

import static com.new1cloud.file.upload.client.bootstrap.handler.LocalCache.chunkFiles;

@Setter
public class FileUploadClient {
    protected final Logger log = LoggerFactory.getLogger(this.getClass());
    /**
     * 分片大小(单位byte)
     */
    public static final long defaultChunkSize = 20 * 1048576L;

    /**
     * 每个客户端处理的上传分片文件数
     */
    public static final Integer chunkFileCountPerClient = 4;

    /**
     * 存在每个Client负责上传的分片文件名列表
     */
    public static final Map<Channel, Set<String>> mapChunkFileNameByClient = new ConcurrentHashMap<>();

    /**
     * 服务端地址
     */
    private String serverAddress;
    /**
     * 服务端口
     */
    private int port;

    /**
     * 待发送文件的绝对路径(支持多个文件的发送)
     */
    private List<String> sourceFilePaths;

    /**
     * 上传的远端目录
     */
    private String uploadDir;

    /**
     * 用于启动Client的线程池
     */
    private ExecutorService clientExecutorService;

    /**
     * 上传进度监听器
     */
    private UploadProgressListener uploadProgressListener;

    //对chunkFiles进行分组
    private List<FileUploadBasicInfo> groupFilesBasicInfo;

    //计数器
    private CountDownLatch latch;

    public FileUploadClient(String serverAddress, int port, List<String> sourceFilePaths, String uploadDir) {
        this.serverAddress = serverAddress;
        this.port = port;
        this.sourceFilePaths = sourceFilePaths;
        this.uploadDir = uploadDir;

        //cpu核心数
        int availableProcessors = Runtime.getRuntime().availableProcessors();
        this.clientExecutorService = new ThreadPoolExecutor(availableProcessors - 1,
                availableProcessors * 2 - 1,
                0,
                TimeUnit.MILLISECONDS,
                new LinkedBlockingQueue<>(1000)
        );

        //缓存相关的数据
        //简单地切分文件,and cache data
        sliceAndCacheFiles(this.sourceFilePaths, this.uploadDir);
        //对chunkFiles进行分组
        this.groupFilesBasicInfo = splitFileChunkInfo(chunkFileCountPerClient);
        this.latch = new CountDownLatch(groupFilesBasicInfo.size());

        Runtime.getRuntime().addShutdownHook(new Thread(() -> {
            if(Objects.nonNull(this.clientExecutorService)) {
                log.info("即将开始关闭clientExecutorService.");
                this.clientExecutorService.shutdown();
            }
        }));
    }

    /**
     * 获取UploadProgressListener， 确保该字段是非空的
     * @return UploadProgressListener
     */
    private UploadProgressListener gotUploadProgressListener(){
        return Objects.nonNull(uploadProgressListener) ? uploadProgressListener :
                var1 -> log.info("文件({})的【上传进度】: {}%", var1.getFileName(), var1.getTransferPercentage());
    }

    public void run() {
        final long startTime = System.currentTimeMillis();
        if(Objects.isNull(sourceFilePaths) || sourceFilePaths.isEmpty()) {
            log.error("sourceFilePaths不能为空.");
            return;
        }

        EventLoopGroup group = new NioEventLoopGroup(groupFilesBasicInfo.size());
        try {
            Bootstrap b = new Bootstrap();
            b.group(group)
                .channel(NioSocketChannel.class)
                .option(ChannelOption.TCP_NODELAY, true)
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel ch) {
                        ChannelPipeline p = ch.pipeline();
                        p.addLast(new LengthFieldPrepender(4));
                        p.addLast(new ObjectEncoder());
                        p.addLast(new LengthFieldBasedFrameDecoder(64 * 1024,0,4,0,4));
                        p.addLast(new ObjectDecoder(200 * 1024 * 1024, ClassResolvers.weakCachingConcurrentResolver(this.getClass().getClassLoader())));
                        p.addLast(new FileUploadClientHandler(gotUploadProgressListener(), latch));
//                        p.addLast(new ChunkFileUploadStatusHandler(gotUploadProgressListener()));
                    }
                });

            //并发启动多个Client进行文件上传
            int clientIndex = 1;
            for(FileUploadBasicInfo fileBasicInfo : groupFilesBasicInfo) {
                String clientName = "客户端_" + clientIndex;
                this.clientExecutorService.submit(() -> {
                    Channel channel = null;
                    try {
//                        log.debug("{}需要上传的分片文件信息: {}", clientName, fileBasicInfo.getNeedSendChunkFiles().getChunkFiles());
                        channel = new FileUploadSubClient(serverAddress, port, Collections.singletonList(fileBasicInfo), b, clientName)
                                .execute();
                    } finally {
                        //当channel关闭之后才算这个Client上传完所有的分片文件
//                        if(Objects.nonNull(channel) && !channel.isOpen()) {
//                            log.info("CountDownLatch将会减一!");
//                            this.latch.countDown();
//                        }
                    }
                });
                clientIndex++;
            }
            this.latch.await();
        } catch (Exception e) {
            throw new RuntimeException(e);
        } finally {
            log.info("执行完成，客户端将退出.");
            final long endTime = System.currentTimeMillis();
            log.info("本次上传文件所花费的时间秒: {}", (endTime - startTime) / 1000);
            group.shutdownGracefully();
        }
    }

    /**
     * 批量将多个文件进行分片拆分
     * @param sourceFilePaths 文件路径列表
     */
    private void sliceAndCacheFiles(List<String> sourceFilePaths, String uploadDir) {
        sourceFilePaths.stream().map(path -> this.sliceFile(path, uploadDir))
                .forEach(fileChunkInfo -> chunkFiles.put(fileChunkInfo.getFileHash(), fileChunkInfo));
    }

    /**
     * 对{@link LocalCache#chunkFiles}里的FileChunkInfos进行拆分
     * @param batchSize 每个子集的元素个数
     * @return 子集合
     */
    private List<FileUploadBasicInfo> splitFileChunkInfo(Integer batchSize){
        return chunkFiles.values().stream().map(fileChunkInfo -> {
            ChunkFile chunkFile = fileChunkInfo.getChunkFiles().get(0);
            List<FileChunkInfo> fileChunkInfos = FileChunkInfo.splitByChunkFiles(fileChunkInfo, batchSize);
            return fileChunkInfos.stream().map(subFileChunkInfo -> FileUploadBasicInfo.builder().fileHash(fileChunkInfo.getFileHash())
                    .fileName(chunkFile.getFileName())
                    .fileSize(chunkFile.getFileSize())
                    .needSendChunkFiles(subFileChunkInfo)
                    .uploadPath(uploadDir)
                    .build()).collect(Collectors.toList());
        }).flatMap(Collection::stream).collect(Collectors.toList());
    }

    /**
     * 将单个文件进行分片
     * @param sourceFilePath 源文件的绝对路径
     * @param uploadDir 上传的远端目录
     * @return 分片文件列表
     */
    private FileChunkInfo sliceFile(String sourceFilePath, String uploadDir) {
        File sourceFile = new File(sourceFilePath);
        String fileHash = FileHashGenerate.genHashByAttributes(sourceFile);
        String fileName = sourceFile.getName();
        long fileSize = sourceFile.length();
        // 计算文件总分片数
        int chunkNum = (int) Math.ceil(1.0 * fileSize / defaultChunkSize);
        List<ChunkFile> chunkFiles = new ArrayList<>(chunkNum);
        long offset = 0;
        for(int i = 1; i <= chunkNum; i++) {
            ChunkFile chunkFile = new ChunkFile(fileName, fileHash, fileSize, offset, i+"", null, i == chunkNum ? 1 : 0, uploadDir);
            chunkFiles.add(chunkFile);
            log.info("【逻辑切分文件】({})得到的ChunkFile: {}", sourceFilePath, chunkFile);
            // 更新偏移量
            offset += defaultChunkSize;
        }
        return FileChunkInfo.builder().chunkFiles(chunkFiles).filePath(sourceFilePath).fileHash(fileHash)
                .build();
    }

}
